In [1]:
import os
os.environ['KERAS_BACKEND']='tensorflow' # can choose theano, tensorflow, cntk
os.environ['THEANO_FLAGS']='floatX=float32,device=cuda,optimizer=fast_run,dnn.library_path=/usr/lib'
import keras.backend as K
K.set_image_data_format('channels_last')
channel_axis = -1
In [3]:
from keras.models import Sequential, Model
from keras.layers import Conv2D, ZeroPadding2D, BatchNormalization, Input, Dropout
from keras.layers import Conv2DTranspose, Reshape, Activation, Cropping2D, Flatten
from keras.layers import Concatenate
from keras.layers.advanced_activations import LeakyReLU
from keras.activations import relu
from keras.initializers import RandomNormal
In [4]:
# Weights initializations
# bias are initailized as 0
#conv 웨이트 초기화
def __conv_init(a):
print("conv_init", a)
k = RandomNormal(0, 0.02)(a) # for convolution kernel
k.conv_weight = True
return k
conv_init = RandomNormal(0, 0.02)
gamma_init = RandomNormal(1., 0.02) # for batch normalization
In [5]:
# Basic discriminator
def conv2d(f, *a, **k):
return Conv2D(f, kernel_initializer = conv_init, *a, **k)
def batchnorm():
#batchnormalization
return BatchNormalization(momentum=0.9, axis=channel_axis, epsilon=1.01e-5,
gamma_initializer = gamma_init)
def BASIC_D(nc_in, nc_out, ndf, max_layers=3):
"""DCGAN_D(nc, ndf, max_layers=3)
nc: channels
ndf: filters of the first layer
max_layers: max hidden layers
"""
input_a, input_b = Input(shape=(None, None, nc_in)), Input(shape=(None, None, nc_out))
#두개의 인풋을 concat
_ = Concatenate(axis=channel_axis)([input_a, input_b])
#CNN Layer
_ = conv2d(ndf, kernel_size=4, strides=2, padding="same", name = 'First') (_)
_ = LeakyReLU(alpha=0.2)(_)
#layer를 max_layer 만큼 중첩해서 쌓음
for layer in range(1, max_layers):
out_feat = ndf * min(2**layer, 8)
_ = conv2d(out_feat, kernel_size=4, strides=2, padding="same",
use_bias=False, name = 'pyramid.{0}'.format(layer)
) (_)
_ = batchnorm()(_, training=1)
_ = LeakyReLU(alpha=0.2)(_)
out_feat = ndf*min(2**max_layers, 8)
_ = ZeroPadding2D(1)(_)
_ = conv2d(out_feat, kernel_size=4, use_bias=False, name = 'pyramid_last') (_)
_ = batchnorm()(_, training=1)
_ = LeakyReLU(alpha=0.2)(_)
# final layer
_ = ZeroPadding2D(1)(_)
_ = conv2d(1, kernel_size=4, name = 'final'.format(out_feat, 1),
activation = "sigmoid") (_)
return Model(inputs=[input_a, input_b], outputs=_)
In [6]:
def UNET_G(isize, nc_in=3, nc_out=3, ngf=64, fixed_input_size=True):
max_nf = 8*ngf
def block(x, s, nf_in, use_batchnorm=True, nf_out=None, nf_next=None):
print("block",x,s,nf_in, use_batchnorm, nf_out, nf_next)
assert s>=2 and s%2==0
if nf_next is None:
nf_next = min(nf_in*2, max_nf)
if nf_out is None:
nf_out = nf_in
x = conv2d(nf_next, kernel_size=4, strides=2, use_bias=(not (use_batchnorm and s>2)),
padding="same", name = 'conv_{0}'.format(s)) (x)
if s>2:
if use_batchnorm:
x = batchnorm()(x, training=1)
x2 = LeakyReLU(alpha=0.2)(x)
x2 = block(x2, s//2, nf_next)
x = Concatenate(axis=channel_axis)([x, x2])
x = Activation("relu")(x)
x = Conv2DTranspose(nf_out, kernel_size=4, strides=2, use_bias=not use_batchnorm,
kernel_initializer = conv_init,
name = 'convt.{0}'.format(s))(x)
x = Cropping2D(1)(x)
if use_batchnorm:
x = batchnorm()(x, training=1)
if s <=8:
x = Dropout(0.5)(x, training=1)
return x
s = isize if fixed_input_size else None #image size
_ = inputs = Input(shape=(s, s, nc_in)) #input layer
_ = block(_, isize, nc_in, False, nf_out=nc_out, nf_next=ngf)
_ = Activation('tanh')(_)
return Model(inputs=inputs, outputs=[_])
In [7]:
nc_in = 3 #채널 수
nc_out = 3
ngf = 64 #필터 수
ndf = 64
lamda = 10
loadSize = 286
imageSize = 256
batchSize = 1
lrD = 2e-4
lrG = 2e-4
In [8]:
netD = BASIC_D(nc_in, nc_out, ndf)
netD.summary()
In [9]:
from IPython.display import SVG
from keras.utils.vis_utils import model_to_dot
netG = UNET_G(imageSize, nc_in, nc_out, ngf)
SVG(model_to_dot(netG, show_shapes=True).create(prog='dot', format='svg'))
netG.summary()
In [10]:
def extract_patches(images, sub_patch_dim):
"""
Cuts images into k subpatches
Each kth cut as the kth patches for all images
ex: input 3 images [im1, im2, im3]
output [[im_1_patch_1, im_2_patch_1], ... , [im_n-1_patch_k, im_n_patch_k]]
:param images: array of Images (num_images, im_channels, im_height, im_width)
:param sub_patch_dim: (height, width) ex: (30, 30) Subpatch dimensions
:return:
"""
im_height, im_width = images.shape[1:3]
print(images.shape)
patch_height, patch_width = sub_patch_dim
# list out all xs ex: 0, 29, 58, ...
x_spots = range(0, im_width, patch_width)
# list out all ys ex: 0, 29, 58
y_spots = range(0, im_height, patch_height)
all_patches = []
for y in y_spots:
for x in x_spots:
# indexing here is cra
# images[num_images, num_channels, width, height]
# this says, cut a patch across all images at the same time with this width, height
image_patches = images[:, y: y+patch_height, x: x+patch_width, :]
all_patches.append(np.float32(image_patches))
return all_patches
In [11]:
from keras.optimizers import RMSprop, SGD, Adam
In [ ]:
def tensor_mean(A):
sum_tensor = 0
for t in A:
sum_tensor += t
return (sum_tensor/len(A))
def make_patch_Tensor(A):
patch_list = []
for ptc in K.tf.split(A,4,axis=1):
patch_list.extend(K.tf.split(ptc,4,axis=2))
return patch_list
In [14]:
real_A = netG.input #real image
fake_B = netG.output #fake image
netG_generate = K.function([real_A], [fake_B])#img generator
real_B = K.tf.placeholder(dtype="float32",shape=(None,None,None,3)) #discriminator의 인풋
real_A_patches = make_patch_Tensor(real_A)
real_B_patches = make_patch_Tensor(real_B)
fake_B_patches = make_patch_Tensor(fake_B)
"""
output_D_real = netD([real_A, real_B]) #D의 real 아웃풋
output_D_fake = netD([real_A, fake_B]) #D의 fake 아웃풋
"""
output_D_real = [netD([a, b]) for a,b in zip(real_A_patches,real_B_patches)]
output_D_fake = [netD([a, b]) for a,b in zip(real_A_patches,fake_B_patches)]
output_D_real = tensor_mean(output_D_real)
output_D_fake = tensor_mean(output_D_fake)
In [15]:
loss_fn = lambda output, target : -K.mean(K.log(output+1e-12)*target+K.log(1-output+1e-12)*(1-target))
loss_D_real = loss_fn(output_D_real, K.ones_like(output_D_real))
loss_D_fake = loss_fn(output_D_fake, K.zeros_like(output_D_fake)) #가짜 분별 학습
loss_G_fake = loss_fn(output_D_fake, K.ones_like(output_D_fake)) #미스 Leading?
loss_L1 = K.mean(K.abs(fake_B-real_B))
In [16]:
loss_D = loss_D_real +loss_D_fake
training_updates = Adam(lr=lrD, beta_1=0.5).get_updates(netD.trainable_weights,[],loss_D)
netD_train = K.function([real_A, real_B],[loss_D/2.0], training_updates)
In [17]:
loss_G = loss_G_fake + 100 * loss_L1
training_updates = Adam(lr=lrG, beta_1=0.5).get_updates(netG.trainable_weights,[], loss_G)
netG_train = K.function([real_A, real_B], [loss_G_fake, loss_L1], training_updates)
In [18]:
from PIL import Image
import numpy as np
import glob
from random import randint, shuffle
def load_data(file_pattern):
return glob.glob(file_pattern)
# 두장이 concat된 사진을 두장으로 나누는 부분
def read_image(fn, direction=0):
im = Image.open(fn)
im = im.resize( (loadSize*2, loadSize), Image.BILINEAR )
arr = np.array(im)/255.0*2-1
w1,w2 = (loadSize-imageSize)//2,(loadSize+imageSize)//2
h1,h2 = w1,w2
imgA = arr[h1:h2, loadSize+w1:loadSize+w2, :]
imgB = arr[h1:h2, w1:w2, :]
if direction==0:
return imgA, imgB
else:
return imgB,imgA
direction = 1
trainAB = load_data('./cards_ab/train/*.jpg')
valAB = load_data('./cards_ab/val/*.jpg')
assert len(trainAB) and len(valAB)
In [19]:
def minibatch(dataAB, batchsize, direction=0):
length = len(dataAB)
epoch = i = 0
tmpsize = None
while True:
size = tmpsize if tmpsize else batchsize #batchsize
if i+size > length:
shuffle(dataAB)
i = 0
epoch+=1
dataA = []
dataB = []
for j in range(i,i+size):
imgA,imgB = read_image(dataAB[j], direction)#img 두 장을 받아서 list에 넣어줌
dataA.append(imgA)
dataB.append(imgB)
dataA = np.float32(dataA)
dataB = np.float32(dataB)
i+=size
tmpsize = yield epoch, dataA, dataB
In [20]:
from IPython.display import display
def showX(X, rows=1):
assert X.shape[0]%rows == 0
int_X = ( (X+1)/2.0*255).clip(0,255).astype('uint8')
int_X = int_X.reshape(-1,imageSize,imageSize, 3)
int_X = int_X.reshape(rows, -1, imageSize, imageSize,3).swapaxes(1,2).reshape(rows*imageSize,-1, 3)
display(Image.fromarray(int_X))
In [21]:
train_batch = minibatch(trainAB, 6, direction=direction)
_, trainA, trainB = next(train_batch)
showX(trainA)
showX(trainB)
del train_batch, trainA, trainB
In [22]:
def netG_gen(A):
return np.concatenate([netG_generate([A[i:i+1]])[0] for i in range(A.shape[0])], axis=0)
In [23]:
import time
from IPython.display import clear_output
t0 = time.time()
niter = 50 #epoch
gen_iterations = 0
errL1 = epoch = errG = 0
errL1_sum = errG_sum = errD_sum = 0
display_iters = 500
val_batch = minibatch(valAB, 6, direction)
train_batch = minibatch(trainAB, batchSize, direction)
while epoch < niter:
epoch, trainA, trainB = next(train_batch)
errD, = netD_train([trainA, trainB])
errD_sum +=errD
errG, errL1 = netG_train([trainA, trainB])
errG_sum += errG
errL1_sum += errL1
gen_iterations+=1
if gen_iterations%display_iters==0:
if gen_iterations%(5*display_iters)==0:
clear_output()
print('[%d/%d][%d] Loss_D: %f Loss_G: %f loss_L1: %f'
% (epoch, niter, gen_iterations, errD_sum/display_iters, errG_sum/display_iters, errL1_sum/display_iters), time.time()-t0)
_, valA, valB = train_batch.send(6)
fakeB = netG_gen(valA)
showX(np.concatenate([valA, valB, fakeB], axis=0), 3)
errL1_sum = errG_sum = errD_sum = 0
_, valA, valB = next(val_batch)
fakeB = netG_gen(valA)
showX(np.concatenate([valA, valB, fakeB], axis=0), 3)
In [63]:
print(len(patch_B))
In [24]:
_, valA, valB = val_batch.send(6)
fakeB = netG_gen(valA)
showX(np.concatenate([valA, valB, fakeB], axis=0), 3)
errL1_sum = errG_sum = errD_sum = 0
_, valA, valB = next(val_batch)
fakeB = netG_gen(valA)
showX(np.concatenate([valA, valB, fakeB], axis=0), 3)
In [25]:
def read_single_image(fn):
im = Image.open(fn)
im = im.resize( (loadSize, loadSize), Image.BILINEAR )
arr = np.array(im)/255.0*2-1
w1,w2 = (loadSize-imageSize)//2,(loadSize+imageSize)//2
h1,h2 = w1,w2
img = arr[h1:h2, w1:w2, :]
return img
In [26]:
# 0~636
max_idx = 636
src_dir = './cards_ab/test_in/'
dst_dir = './cards_ab/test_out/'
for idx in range(max_idx+1):
data = []
src_path = src_dir + str(idx) + '.jpg'
dst_path = dst_dir + str(idx) + '.jpg'
src_img = read_single_image(src_path)
data.append(src_img)
data = np.float32(data)
fake = netG_gen(data)
int_X = ( (fake[0]+1)/2.0*255).clip(0,255).astype('uint8')
int_X = int_X.reshape(-1,imageSize,imageSize, 3)
int_X = int_X.reshape(1, -1, imageSize, imageSize,3).swapaxes(1,2).reshape(1*imageSize,-1, 3)
dst_img = Image.fromarray(int_X)
dst_img.save(dst_path, quality=90)
In [31]:
a = [1,23,4,5]
np.mean(a)
Out[31]:
In [27]:
netD.save_weights("Discriminator.h5")
netG.save_weights("Generator.h5")
In [ ]: